/*
 * Licensed to Crate.io GmbH ("Crate") under one or more contributor
 * license agreements.  See the NOTICE file distributed with this work for
 * additional information regarding copyright ownership.  Crate licenses
 * this file to you under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.  You may
 * obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  See the
 * License for the specific language governing permissions and limitations
 * under the License.
 *
 * However, if you have executed another commercial license agreement
 * with Crate these terms will supersede the license and you may use the
 * software solely pursuant to the terms of the relevant commercial agreement.
 */

package io.crate.sql.parser;

import static java.util.Objects.requireNonNull;

import java.util.EnumSet;
import java.util.List;
import java.util.Locale;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;

import org.antlr.v4.runtime.BaseErrorListener;
import org.antlr.v4.runtime.CharStreams;
import org.antlr.v4.runtime.CommonToken;
import org.antlr.v4.runtime.CommonTokenStream;
import org.antlr.v4.runtime.ParserRuleContext;
import org.antlr.v4.runtime.RecognitionException;
import org.antlr.v4.runtime.Recognizer;
import org.antlr.v4.runtime.Token;
import org.antlr.v4.runtime.atn.ParserATNSimulator;
import org.antlr.v4.runtime.atn.PredictionContextCache;
import org.antlr.v4.runtime.atn.PredictionMode;
import org.antlr.v4.runtime.dfa.DFA;
import org.antlr.v4.runtime.misc.Pair;
import org.antlr.v4.runtime.misc.ParseCancellationException;
import org.antlr.v4.runtime.tree.TerminalNodeImpl;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.jetbrains.annotations.Nullable;
import org.jetbrains.annotations.VisibleForTesting;

import io.crate.sql.parser.antlr.SqlBaseLexer;
import io.crate.sql.parser.antlr.SqlBaseParser;
import io.crate.sql.parser.antlr.SqlBaseParserBaseListener;
import io.crate.sql.tree.Expression;
import io.crate.sql.tree.MultiStatement;
import io.crate.sql.tree.Node;
import io.crate.sql.tree.Statement;

public class SqlParser {

    private static final Logger LOGGER = LogManager.getLogger(SqlParser.class);
    private static final AtomicReference<DFA[]> DFA_CACHE = new AtomicReference<>(newDecisionToDFA());
    private static final AtomicReference<PredictionContextCache> CONTEXT_CACHE = new AtomicReference<>(new PredictionContextCache());

    private static DFA[] newDecisionToDFA() {
        DFA[] decisionToDFA = new DFA[SqlBaseParser._ATN.getNumberOfDecisions()];
        for (int i = 0; i < decisionToDFA.length; i++) {
            decisionToDFA[i] = new DFA(SqlBaseParser._ATN.getDecisionState(i), i);
        }
        return decisionToDFA;
    }

    private static final BaseErrorListener ERROR_LISTENER = new BaseErrorListener() {
        @Override
        public void syntaxError(Recognizer<?, ?> recognizer, Object offendingSymbol, int line,
                                int charPositionInLine, String message, RecognitionException e) {
            throw new ParsingException(message, e, line, charPositionInLine);
        }
    };

    public static final SqlParser INSTANCE = new SqlParser();

    private final EnumSet<IdentifierSymbol> allowedIdentifierSymbols;

    public SqlParser() {
        this(new SqlParserOptions().allowIdentifierSymbol(IdentifierSymbol.AT_SIGN));
    }

    SqlParser(SqlParserOptions options) {
        requireNonNull(options, "options is null");
        allowedIdentifierSymbols = EnumSet.copyOf(options.getAllowedIdentifierSymbols());
    }

    public static Statement createStatement(String sql) {
        return INSTANCE.generateStatement(sql);
    }

    public static List<Statement> createStatementsForSimpleQuery(String sql, Function<String, Object> stringLiteralParser) {
        return ((MultiStatement) INSTANCE.invokeParser(
                "statements",
                sql,
                SqlBaseParser::statements,
                str -> (Expression) INSTANCE.invokeParser(
                        "parameterOrLiteral",
                        String.format(Locale.ENGLISH, "%s", stringLiteralParser.apply(str)),
                        SqlBaseParser::parameterOrLiteral,
                        null
                )
        )).statements();
    }

    private Statement generateStatement(String sql) {
        return (Statement) invokeParser("statement", sql, SqlBaseParser::singleStatement, null);
    }

    public static Expression createExpression(String expression) {
        return INSTANCE.generateExpression(expression);
    }

    private Expression generateExpression(String expression) {
        return (Expression) invokeParser("expression", expression, SqlBaseParser::singleExpression, null);
    }

    private Node invokeParser(String name,
                              String sql,
                              Function<SqlBaseParser, ParserRuleContext> parseFunction,
                              @Nullable Function<String, Expression> parseStringLiteral) {
        try {
            SqlBaseLexer lexer = new SqlBaseLexer(new CaseInsensitiveStream(CharStreams.fromString(sql, name)));
            CommonTokenStream tokenStream = new CommonTokenStream(lexer);
            SqlBaseParser parser = new SqlBaseParser(tokenStream);
            parser.setInterpreter(new ParserATNSimulator(
                parser,
                parser.getATN(),
                DFA_CACHE.get(),
                CONTEXT_CACHE.get()
            ));

            parser.addParseListener(new PostProcessor());

            lexer.removeErrorListeners();
            lexer.addErrorListener(ERROR_LISTENER);

            parser.removeErrorListeners();
            parser.addErrorListener(ERROR_LISTENER);
            parser.setErrorHandler(new ExceptionErrorStrategy());

            ParserRuleContext tree;
            try {
                // first, try parsing with potentially faster SLL mode
                parser.getInterpreter().setPredictionMode(PredictionMode.SLL);
                tree = parseFunction.apply(parser);
            } catch (ParseCancellationException ex) {
                // if we fail, parse with LL mode
                tokenStream.seek(0); // rewind input stream
                parser.reset();

                parser.getInterpreter().setPredictionMode(PredictionMode.LL);
                tree = parseFunction.apply(parser);
            }

            return new AstBuilder(parseStringLiteral).visit(tree);
        } catch (StackOverflowError e) {
            throw new ParsingException(name + " is too large (stack overflow while parsing)");
        }
    }

    /// Clears the DFA and PredictionContextCache if there are many entries cached
    public static void clearCaches() {
        // Thresholds here are to ensure we don't start thrashing the cache on OOM pressure
        // We only want to clear it if the DFA cache or prediction cache has a significant size.
        //
        // The value was taken by running TestStatementBuilder and measuring the size:
        // There are roughly 4500 DFA entries which require about 2.5 MB of memory.
        // And 48246 prediction cache entries (amounting to 1.54 MB @ 32 byte each
        // (with compressedOps 24 byte each))
        clearCaches(5_000, 50_000);
    }

    @VisibleForTesting
    static void clearCaches(int dfaThreshold, int predictionThreshold) {
        DFA[] dfas = DFA_CACHE.get();
        long totalDfaEntries = 0;
        for (var dfa : dfas) {
            totalDfaEntries += dfa.states.size();
        }
        if (totalDfaEntries > dfaThreshold) {
            DFA_CACHE.set(newDecisionToDFA());
            LOGGER.info("Clearing SqlParser DFA cache. Had {} entries", totalDfaEntries);
        }
        PredictionContextCache predictionContextCache = CONTEXT_CACHE.get();
        int predictionCacheSize = predictionContextCache.size();
        if (predictionCacheSize > predictionThreshold) {
            CONTEXT_CACHE.set(new PredictionContextCache());
            LOGGER.info("Clearing SqlParser PredictionContextCache. Had {} entries", predictionCacheSize);
        }
    }

    static DFA[] dfas() {
        return DFA_CACHE.get();
    }

    static int contextCacheSize() {
        return CONTEXT_CACHE.get().size();
    }

    private class PostProcessor extends SqlBaseParserBaseListener {
        @Override
        public void exitUnquotedIdentifier(SqlBaseParser.UnquotedIdentifierContext context) {
            String identifier = context.IDENTIFIER().getText();
            for (IdentifierSymbol identifierSymbol : EnumSet.complementOf(allowedIdentifierSymbols)) {
                char symbol = identifierSymbol.getSymbol();
                if (identifier.indexOf(symbol) >= 0) {
                    throw new ParsingException("identifiers must not contain '"
                        + identifierSymbol.getSymbol() + "'", null, context.IDENTIFIER().getSymbol().getLine(),
                        context.IDENTIFIER().getSymbol().getCharPositionInLine());
                }
            }
        }

        @Override
        public void exitBackQuotedIdentifier(SqlBaseParser.BackQuotedIdentifierContext context) {
            Token token = context.BACKQUOTED_IDENTIFIER().getSymbol();
            throw new ParsingException(
                "backquoted identifiers are not supported; use double quotes to quote identifiers",
                null,
                token.getLine(),
                token.getCharPositionInLine());
        }

        @Override
        public void exitDigitIdentifier(SqlBaseParser.DigitIdentifierContext context) {
            Token token = context.DIGIT_IDENTIFIER().getSymbol();
            throw new ParsingException(
                "identifiers must not start with a digit; surround the identifier with double quotes",
                null,
                token.getLine(),
                token.getCharPositionInLine());
        }

        @Override
        public void exitNonReserved(SqlBaseParser.NonReservedContext context) {
            // replace nonReserved words with IDENT tokens
            context.getParent().removeLastChild();

            Token token = (Token) context.getChild(0).getPayload();
            context.getParent().addChild(new TerminalNodeImpl(
                new CommonToken(
                    new Pair<>(token.getTokenSource(), token.getInputStream()),
                    SqlBaseLexer.IDENTIFIER,
                    token.getChannel(),
                    token.getStartIndex(),
                    token.getStopIndex())
                )
            );
        }
    }
}
